package kycadminrepo

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"code.bydev.io/cht/customer/kyc_admin_service/internal/constant"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/config"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/helper"
	kae_dto "code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycadminevent/dto"
	kar_dto "code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycadminrepo/dto"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycadminrepo/o_kyc_action_state"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycadminrepo/o_kyc_action_state_log"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycprovider"
	kp_dto "code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycprovider/dto"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycprovider/provider"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycprovider/provider/sumsub"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/monitor"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/utils/timeutil"
	"code.bydev.io/frameworks/byone/core/logc"
	"code.bydev.io/frameworks/byone/core/logx"
	"code.bydev.io/frameworks/byone/kafka"
	"emperror.dev/errors"
	"git.bybit.com/svc/go/pkg/bdal"
	"git.bybit.com/svc/go/pkg/bdal/bqb"
	"git.bybit.com/svc/go/pkg/bdebug"
	"git.bybit.com/svc/go/pkg/bstd"
	"git.bybit.com/svc/go/pkg/bzap"
	kycpb "git.bybit.com/svc/stub/pkg/pb/api/kyc"
	"git.bybit.com/svc/stub/pkg/pb/enums/ekycadmin"
	"git.bybit.com/svc/stub/pkg/pb/enums/kyc"
	"github.com/google/uuid"
	jsoniter "github.com/json-iterator/go"
	"github.com/opentracing/opentracing-go"
	_ "go.elastic.co/apm/module/apmsql/mysql"
	"go.uber.org/zap"
)

const (
	StrategyInsert          string = "INSERT"          // 插入新纪录
	StrategyUpdate          string = "UPDATE"          // 更新上一条记录
	StrategySmbIcrJmoUpd    string = "SMBICRJMOUPD"    // sumsub：更新 jumio：workflowID 一样？y:更新；n:插入
	StrategySmbIcrJmoUpdAlt string = "SMBICRJMOUPDALT" // sumsub：更新 jumio：workflowID 一样？y:更新；n:插入 +告警
	KycLv1                         = 1                 // kyc level1 标识
	KycLv2                         = 2                 // kyc level1 标识
	KycResultSuccess               = 1                 // kyc服务商认证结果为成功
	KycActionTopic                 = "cht.kyc-service-public.kyc.action"
)

var getIdempotentSQL = o_kyc_action_state.Select().WhereC(func(wb *bqb.WhereBuilder) {
	o_kyc_action_state.FldMemberId.EQ(wb)
	o_kyc_action_state.FldSiteId.EQ(wb)
	o_kyc_action_state.FldChannel.EQ(wb)
	o_kyc_action_state.FldLevel.EQ(wb)
	o_kyc_action_state.FldSerialType.EQ(wb)
}).OrderByC(o_kyc_action_state.FldProviderCreateAt, bqb.DESC).LimitC(1).SQL()

var getIdempotentWithRecordIDSQL = o_kyc_action_state.Select().WhereC(func(wb *bqb.WhereBuilder) {
	o_kyc_action_state.FldMemberId.EQ(wb)
	o_kyc_action_state.FldSiteId.EQ(wb)
	o_kyc_action_state.FldChannel.EQ(wb)
	o_kyc_action_state.FldLevel.EQ(wb)
	o_kyc_action_state.FldSerialType.EQ(wb)
	o_kyc_action_state.FldRecordId.EQ(wb)
}).OrderByC(o_kyc_action_state.FldProviderCreateAt, bqb.DESC).LimitC(1).SQL()

var getIdempotentNoLvlSQL = o_kyc_action_state.Select().WhereC(func(wb *bqb.WhereBuilder) {
	o_kyc_action_state.FldMemberId.EQ(wb)
	o_kyc_action_state.FldSiteId.EQ(wb)
	o_kyc_action_state.FldChannel.EQ(wb)
	o_kyc_action_state.FldSerialType.EQ(wb)
}).OrderByC(o_kyc_action_state.FldId, bqb.DESC).LimitC(1).SQL()

// GetIdempotentID generate an idempotent id
func (r *repo) GetIdempotentID(ctx context.Context, current *kar_dto.GetIdempotentCondition) (string, bool, bool, error) {
	spanPl, _ := opentracing.StartSpanFromContext(ctx, "GetIdempotentID")
	defer spanPl.Finish()

	pool, err := r.kycDB.DBDefaultMasterNodePool(ctx)
	if err != nil {
		return "", false, false, err
	}

	var record *o_kyc_action_state.Record

	if current.Level == 0 && current.Action == ekycadmin.KYCAdminAction_ACTION_RESET {
		record, err = o_kyc_action_state.ScanRecordNil(getIdempotentNoLvlSQL.QueryArgs(ctx, pool, bqb.Args{
			o_kyc_action_state.FldMemberId.Param():   current.MemberID,
			o_kyc_action_state.FldSiteId.Param():     current.SiteID,
			o_kyc_action_state.FldChannel.Param():    current.Channel,
			o_kyc_action_state.FldSerialType.Param(): current.SerialType,
		}))
	} else {
		if current.RecordID != "" && current.SerialType == 1 {
			record, err = o_kyc_action_state.ScanRecordNil(getIdempotentWithRecordIDSQL.QueryArgs(ctx, pool, bqb.Args{
				o_kyc_action_state.FldMemberId.Param():   current.MemberID,
				o_kyc_action_state.FldSiteId.Param():     current.SiteID,
				o_kyc_action_state.FldChannel.Param():    current.Channel,
				o_kyc_action_state.FldLevel.Param():      current.Level,
				o_kyc_action_state.FldSerialType.Param(): current.SerialType,
				o_kyc_action_state.FldRecordId.Param():   current.RecordID,
			}))
		} else {
			record, err = o_kyc_action_state.ScanRecordNil(getIdempotentSQL.QueryArgs(ctx, pool, bqb.Args{
				o_kyc_action_state.FldMemberId.Param():   current.MemberID,
				o_kyc_action_state.FldSiteId.Param():     current.SiteID,
				o_kyc_action_state.FldChannel.Param():    current.Channel,
				o_kyc_action_state.FldLevel.Param():      current.Level,
				o_kyc_action_state.FldSerialType.Param(): current.SerialType,
			}))
		}
	}

	if err != nil {
		return "", false, false, err
	}
	logc.Infow(ctx, "get kyc_action_state record", logc.Field("record", record))
	// 无记录-插入
	if record == nil {
		return uuid.New().String(), true, false, nil
	}
	// 过期消息-更新
	if current.ProviderCreateAt <= record.ProviderCreateAt {
		return record.IdempotentId, false, false, nil
	}
	s := r.getStrategy(ekycadmin.KYCAdminAction(record.Action), current.Action)
	switch s {
	case StrategyInsert:
		return uuid.New().String(), true, false, nil
	case StrategyUpdate:
		return record.IdempotentId, false, false, nil
	case StrategySmbIcrJmoUpd:
		// sumsub：更新
		// jumio：workflowID 一样？y:更新；n:插入
		if kyc.KYCProvider(record.Channel) == kyc.KYCProvider_PROVIDER_SUMSUB || record.RecordId == current.RecordID {
			return record.IdempotentId, false, false, nil
		}
		return uuid.New().String(), true, false, nil
	case StrategySmbIcrJmoUpdAlt:
		// sumsub：更新
		// jumio：workflowID 一样？y:更新；n:插入 +告警
		if kyc.KYCProvider(record.Channel) == kyc.KYCProvider_PROVIDER_SUMSUB || record.RecordId == current.RecordID {
			return record.IdempotentId, false, false, nil
		}
		return uuid.New().String(), true, true, nil
	}

	return uuid.New().String(), true, false, nil
}

func (r *repo) getStrategy(last ekycadmin.KYCAdminAction, current ekycadmin.KYCAdminAction) string {
	// strategyTbl 键: 第一个数字表示上一次事件的状态; 第二个数字表示本次事件的状态
	// 值: 数据的更新或插入的策略
	strategyTbl := map[string]string{
		"0-0": StrategySmbIcrJmoUpd,
		"0-1": StrategyUpdate,
		"0-2": StrategyUpdate,
		"0-3": StrategyUpdate,
		"0-4": StrategyUpdate,
		"1-0": StrategySmbIcrJmoUpdAlt,
		"1-1": StrategySmbIcrJmoUpd,
		"1-2": StrategyUpdate,
		"1-3": StrategyUpdate,
		"1-4": StrategyUpdate,
		"2-0": StrategyUpdate,
		"2-1": StrategyUpdate,
		"2-2": StrategyUpdate,
		"2-3": StrategyUpdate,
		"2-4": StrategyUpdate,
		"3-0": StrategyInsert,
		"3-1": StrategyInsert,
		"3-2": StrategyInsert,
		"3-3": StrategyUpdate,
		"3-4": StrategyUpdate,
		"4-0": StrategyInsert,
		"4-1": StrategyInsert,
		"4-2": StrategyInsert,
		"4-3": StrategyInsert,
		"4-4": StrategyUpdate,
	}
	k := fmt.Sprintf("%d-%d", int(last), int(current))
	if s, ok := strategyTbl[k]; ok {
		return s
	}

	return StrategyInsert
}

func (r *repo) initActionStateRecord(ctx context.Context, data *kar_dto.ActionStateRecord) *o_kyc_action_state.Record {
	adJSON, _ := jsoniter.MarshalToString(kp_dto.KYCAdminAddressDetail{Country: "", State: "", City: "", FullAddress: "", PostalCode: ""})
	adJSONDe, err := helper.EnAES256CodeString(r.secHub, adJSON)
	if err != nil {
		r.logger.Error("EnAES256CodeString adJSON error",
			zap.Any("reqID", helper.GetReqID(ctx)),
			zap.Error(err))
	}

	ddJSON, _ := jsoniter.MarshalToString(kp_dto.KYCAdminDocDetail{DocFront: "", DocBack: "", PersonalNumber: ""})
	ddJSONDe, err := helper.EnAES256CodeString(r.secHub, ddJSON)
	if err != nil {
		r.logger.Error("EnAES256CodeString ddJSONDe error",
			zap.Any("reqID", helper.GetReqID(ctx)),
			zap.Error(err))
	}

	fdJSON, _ := jsoniter.MarshalToString(kp_dto.KYCAdminFaceDetail{FaceImage: "", LivenessImage: "", SimilarityChecked: "", ValidityChecked: "false"})
	fdJSONDe, err := helper.EnAES256CodeString(r.secHub, fdJSON)
	if err != nil {
		r.logger.Error("EnAES256CodeString fdJSONDe error",
			zap.Any("reqID", helper.GetReqID(ctx)),
			zap.Error(err))
	}

	stateRecord := o_kyc_action_state.New()
	stateRecord.MemberId = data.MemberID
	stateRecord.SiteId = data.SiteID
	stateRecord.ExternalUid = data.ExternalUID
	stateRecord.RecordId = data.RecordID
	stateRecord.AccountId = data.AccountID
	stateRecord.VerifyContent = data.VerifyContent
	stateRecord.Level = data.Level
	stateRecord.Channel = data.Channel
	stateRecord.Action = data.Action
	stateRecord.Decision = data.Decision

	stateRecord.ProviderDecision = data.ProviderDecision
	stateRecord.RejectReason = data.RejectReason
	stateRecord.ProviderRejectReason = data.ProviderRejectReason
	stateRecord.DocType = data.DocType
	stateRecord.DocSubtype = data.DocSubtype
	stateRecord.DocNumber = data.DocNumber
	stateRecord.Country = data.Country
	stateRecord.Nationality = data.Nationality
	stateRecord.AddressCountry = data.AddressCountry
	stateRecord.FirstName = data.FirstName

	stateRecord.LastName = data.LastName
	stateRecord.Dob = data.Dob
	stateRecord.IdempotentId = data.IdempotentID
	stateRecord.ReviewTime = data.ReviewTime
	stateRecord.ProviderCreateAt = data.ProviderCreateAt
	stateRecord.DocDetail = ddJSONDe
	stateRecord.IssuingDate = time.Time{}
	stateRecord.ExpireDate = time.Time{}
	stateRecord.FaceDetail = fdJSONDe
	stateRecord.AddressDetail = adJSONDe
	stateRecord.ClientDetail = data.ClientDetail

	tNow := time.Now()
	stateRecord.CreatedAt = tNow
	stateRecord.UpdatedAt = tNow
	stateRecord.SerialType = data.SerialType

	return stateRecord
}

func (r *repo) CreateActionStateRecord(ctx context.Context, session *bdal.TX, record *o_kyc_action_state.Record) error {
	var (
		pool *bdal.DBPool
		err  error
	)
	if session == nil {
		pool, err = r.kycDB.DBDefaultMasterNodePool(ctx)
		if err != nil {
			return err
		}
	}

	if session != nil {
		return record.Insert(ctx, session)
	} else {
		return record.Insert(ctx, pool)
	}
}

func (r *repo) CreateActionStateLogRecord(ctx context.Context, session *bdal.TX, data *kar_dto.ActionStateLogRecord) error {
	var (
		pool *bdal.DBPool
		err  error
	)
	if session == nil {
		pool, err = r.kycDB.DBDefaultMasterNodePool(ctx)
		if err != nil {
			return err
		}
	}

	stateLogRecord := o_kyc_action_state_log.New()
	stateLogRecord.IdempotentId = data.IdempotentID
	stateLogRecord.MemberId = data.MemberID
	stateLogRecord.SiteId = data.SiteID
	stateLogRecord.RecordId = data.RecordID
	stateLogRecord.AccountId = data.AccountID
	stateLogRecord.Channel = data.Channel
	stateLogRecord.Level = data.Level
	stateLogRecord.VerifyContent = data.VerifyContent
	stateLogRecord.Action = data.Action
	stateLogRecord.EventAt = data.EventAt
	stateLogRecord.EventData = data.EventData
	stateLogRecord.ReservedField1 = data.ReservedField1
	stateLogRecord.ReservedField2 = data.ReservedField2

	tNow := time.Now()
	stateLogRecord.CreatedAt = tNow
	stateLogRecord.UpdatedAt = tNow

	if session != nil {
		return stateLogRecord.Insert(ctx, session)
	} else {
		return stateLogRecord.Insert(ctx, pool)
	}
}

// KYC event handler
func (r *repo) ActionStateHandler(ctx context.Context, ntf *kae_dto.KYCEventNotify) (bool, error) {
	var (
		KYCDetail    *kp_dto.KYCAdminDetail
		actionImages []*kp_dto.KYCActionImage
	)
	metrics := monitor.NewMetrics("ActionStateHandler", "")

	eventData, _ := jsoniter.MarshalToString(ntf)
	logc.Infow(ctx, "KYC event notify", logc.Field("eventData", eventData))

	// 证件序列类型(0:第一证件;1:第二证件)
	serialType := int8(ntf.GetSerialType())
	// Get idempotent id
	cond := &kar_dto.GetIdempotentCondition{
		MemberID:         ntf.MemberID,
		SiteID:           ntf.BrokerID,
		Channel:          ntf.Provider,
		ProviderCreateAt: ntf.EventTimeMs,
		Action:           ntf.EventType,
		Level:            int8(ntf.Level),
		RecordID:         ntf.RecordID,
		SerialType:       serialType,
	}
	idempotentID, isNew, isAlert, err := r.GetIdempotentID(ctx, cond)
	if err != nil {
		metrics.AddMetrics(err, "GetIdempotentID")
		logc.Errorw(ctx, "r.GetIdempotentID error", logc.Field("cond", cond), logc.Field("err", err))
		return false, err
	}

	if isAlert {
		defer func() {
			txt := fmt.Sprintf("recordID:%s eventType:%d", ntf.RecordID, ntf.EventType)
			_ = r.teller.KycAlarm(ctx, txt, errors.New("KYC event unexpected"))
		}()
	}

	logc.Infow(ctx, "ActionStateHandler", logc.Field("idempotentID", idempotentID))

	// 在 DB 操作前先获取 KYC 信息
	if ntf.EventType == ekycadmin.KYCAdminAction_ACTION_COMPLETE {
		KYCDetail, actionImages, err = r.getKYCDetail(ctx, idempotentID, ntf)
		if err != nil {
			logc.Errorw(ctx, "r.getKYCDetail error", logc.Field("err", err))
			return false, err
		}
		if KYCDetail == nil {
			logc.Infow(ctx, "r.getKYCDetail nil", logc.Field("notify", ntf))
			return false, nil
		}
		// 判断是为空Lv2调接口更新用户居住地址
		if ntf.Level > KycLv1 && KYCDetail.ProviderDecision == KycResultSuccess {
			_ = r.appendMemberKycApplicant(ctx, ntf, KYCDetail)
		}
	}

	pool, err := r.kycDB.DBDefaultMasterNodePool(ctx)
	if err != nil {
		logc.Errorw(ctx, "r.kycDB.DBDefaultMasterNodePool error", logc.Field("err", err))
		return false, err
	}
	session, txDispose, err := pool.BeginTX(ctx, nil)
	if err != nil {
		logc.Errorw(ctx, "pool.BeginTX error", logc.Field("err", err))
		return false, err
	}
	defer bstd.WithErrorHandler(txDispose, bdebug.Debug)()

	// 前面getKYCDetail耗时太大，以防这段时间有并发写入，重新判断一次
	_, stillNew, _, _ := r.GetIdempotentID(ctx, cond)
	if isNew && !stillNew {
		logc.Errorw(ctx, "Duplicated Request")
		return false, nil
	}

	logc.Infow(ctx, "Is New record")

	// Sumsub PoI as PoA subtype
	docSubType := kp_dto.UndefinedDocType
	if ntf.Level == 2 && ntf.ResultDetail.FinalDecision == 1 &&
		ntf.Provider == kyc.KYCProvider_PROVIDER_SUMSUB && ntf.PoaIDDocType != "" {
		logc.Infow(ctx, "PoaIDDocType", logc.Field("PoaIDDocType", ntf.PoaIDDocType))
		docSubType = provider.GetUnifiedDocType(sumsub.GetDocType(ntf.PoaIDDocType).String(), kp_dto.ProviderSumSub)
	}

	if isNew { // Create state
		cd := &kp_dto.KYCAdminClientDetail{ClientIP: "", From: "", Source: ""}
		if ntf.EventType == ekycadmin.KYCAdminAction_ACTION_INIT {
			cd.From = ntf.ClientDetail.Device
			cd.ClientIP = ntf.ClientDetail.IP
		}
		cdJSON, _ := jsoniter.MarshalToString(cd)

		recordDto := &kar_dto.ActionStateRecord{
			MemberID:             ntf.MemberID,
			SiteID:               int16(ntf.BrokerID),
			ExternalUID:          ntf.ExternalUID,
			RecordID:             ntf.RecordID,
			AccountID:            ntf.AccountID,
			VerifyContent:        "",
			Level:                int8(ntf.Level),
			Channel:              int8(ntf.Provider),
			Action:               int8(ntf.EventType),
			Decision:             0,
			ProviderDecision:     0,
			RejectReason:         "",
			ProviderRejectReason: "",
			DocType:              0,
			DocSubtype:           int8(docSubType),
			DocNumber:            "",
			Country:              "",
			Nationality:          "",
			AddressCountry:       "",
			FirstName:            "",
			LastName:             "",
			Dob:                  "",
			IdempotentID:         idempotentID,
			ReviewTime:           ntf.EventTimeMs,
			ProviderCreateAt:     ntf.EventTimeMs,
			DocDetail:            "",
			IssuingDate:          time.Time{},
			ExpireDate:           time.Time{},
			FaceDetail:           "",
			AddressDetail:        "",
			ClientDetail:         cdJSON,
			SerialType:           serialType,
		}
		record := r.initActionStateRecord(ctx, recordDto)

		switch ntf.EventType {
		case ekycadmin.KYCAdminAction_ACTION_INIT:
			r.appendAccountID(ctx, record, ntf)
		case ekycadmin.KYCAdminAction_ACTION_COMPLETE:
			errC := r.actionCompleteHandler(ctx, record, ntf, KYCDetail)
			if errC != nil {
				logc.Errorw(ctx, "r.actionCompleteHandler error", logc.Field("err", errC))
				return false, errC
			}
		}
		// 处理 businessID
		r.businessIDHandler(record, ntf)

		logc.Infow(ctx, "state record after", logc.Field("record", record))

		actionStateErr := r.CreateActionStateRecord(ctx, session, record)
		_, isBizErr := bdal.IsDuplicateEntryErr(actionStateErr)
		if isBizErr {
			// 重复插入unique的数据，作为正常case处理, 直接放行
			logc.Infow(ctx, "data has been created", logc.Field("idempotentID", idempotentID))
		}
		if actionStateErr != nil {
			return false, actionStateErr
		}
	} else { // Update state
		record, err := r.GetActionByIdempotentID(ctx, idempotentID)
		if err != nil {
			return false, err
		}
		switch ntf.EventType {
		case ekycadmin.KYCAdminAction_ACTION_INIT: // Init
			r.appendAccountID(ctx, record, ntf)
		case ekycadmin.KYCAdminAction_ACTION_CREATE: // Create
			r.actionCreateHandler(ctx, record, ntf)
		case ekycadmin.KYCAdminAction_ACTION_PENDING: // Pending
			r.actionPendingHandler(ctx, record, ntf)
		case ekycadmin.KYCAdminAction_ACTION_COMPLETE: // Complete
			errC := r.actionCompleteHandler(ctx, record, ntf, KYCDetail)
			if errC != nil {
				logc.Errorw(ctx, "r.actionCompleteHandler error", logc.Field("err", errC))
				return false, errC
			}
		case ekycadmin.KYCAdminAction_ACTION_RESET: // Reset
			r.actionResetHandler(ctx, record, ntf)
		}
		if ntf.AccountID != "" {
			record.AccountId = ntf.AccountID
		}
		if ntf.ExternalUID != "" {
			record.ExternalUid = ntf.ExternalUID
		}
		if ntf.RecordID != "" {
			record.RecordId = ntf.RecordID
		}
		record.DocSubtype = int8(docSubType)
		record.UpdatedAt = time.Now()

		// 处理 businessID
		r.businessIDHandler(record, ntf)

		// 更新状态表
		logc.Infow(ctx, "update record", logc.Field("record", record))
		_, err = record.Update(ctx, session)
		if err != nil {
			return false, err
		}
	}
	// 插入日志

	act := int8(ntf.EventType)
	if ntf.EventType == ekycadmin.KYCAdminAction_ACTION_RESET && ntf.IsAmend {
		// KYC 后台自己修改结果
		act = int8(ekycadmin.KYCAdminAction_ACTION_COMPLETE)
	}
	actionStateLogErr := r.CreateActionStateLogRecord(ctx, session, &kar_dto.ActionStateLogRecord{
		IdempotentID:   idempotentID,
		MemberID:       ntf.MemberID,
		SiteID:         int16(ntf.BrokerID),
		RecordID:       ntf.RecordID,
		AccountID:      ntf.AccountID,
		Channel:        int8(ntf.Provider),
		Level:          int8(ntf.Level),
		VerifyContent:  "",
		Action:         act,
		EventAt:        ntf.EventTimeMs,
		EventData:      eventData,
		ReservedField1: "",
		ReservedField2: 0,
	})
	if actionStateLogErr != nil {
		logc.Errorw(ctx, "CreateActionStateLogRecord error", logc.Field("err", err))
		return false, actionStateLogErr
	}

	// 插入图片
	if ntf.EventType == ekycadmin.KYCAdminAction_ACTION_COMPLETE && ntf.ResultDetail.FinalDecision == KycResultSuccess {
		if len(actionImages) > 0 {
			actionImageErr := r.CreateActionImages(ctx, session, actionImages)
			if actionImageErr != nil {
				logc.Errorw(ctx, "CreateActionImages error", logc.Field("err", actionImageErr))
				return false, actionImageErr
			}
		} else {
			logc.Warnw(ctx, "ActionImagesEmptyWarning", logc.Field("actionImages", actionImages), logc.Field("eventData", eventData))
		}
	}

	// 提交事务
	if e := session.Commit(); e != nil {
		logc.Errorw(ctx, "session.Commit error", logc.Field("err", e))
		return false, e
	}

	// 加推消息
	if ntf.EventType == ekycadmin.KYCAdminAction_ACTION_COMPLETE && ntf.Level == KycLv2 &&
		ntf.ResultDetail.FinalDecision == KycResultSuccess {
		errLv2Notify := r.sendLevel2StateComplete(ctx, ntf)
		if errLv2Notify != nil {
			logc.Errorw(ctx, "r.sendLevel2StateComplete error", logc.Field("err", errLv2Notify))
		}
		logc.Infow(ctx, "r.sendLevel2StateComplete success", logc.Field("ntf", ntf))
	}

	// Poa EEA + UK 失败告警
	logc.Infow(ctx, "Poa EEA", logc.Field("ntf", ntf), logc.Field("KYCDetail", KYCDetail))
	if kp_dto.KYCLevel(ntf.Level) == kp_dto.Level2 && kp_dto.ReviewResult(ntf.ResultDetail.FinalDecision).IsReject() &&
		ntf.EventType == ekycadmin.KYCAdminAction_ACTION_COMPLETE && KYCDetail != nil {
		countryList := config.GetPoaAlarmConfig().CountryList
		logc.Infow(ctx, "Poa EEA", logc.Field("countryList", countryList))
		for _, c := range countryList {
			if KYCDetail.Country == c {
				rjtTimes, err := r.GetPoaRejectTimes(ctx, ntf.MemberID, int8(ntf.Provider))
				if err != nil {
					logc.Errorw(ctx, "r.GetPoaRejectTimes error", logc.Field("err", err))
					return false, err
				}
				logc.Infow(ctx, "Poa EEA reject times", logc.Field("rjtTimes", rjtTimes))
				if rjtTimes >= config.GetPoaAlarmConfig().MinTimes {
					tm := time.Unix(0, ntf.EventTimeMs*int64(time.Millisecond))
					reason := ""
					if KYCDetail.ProviderRejectReason != "" {
						labelSlice := strings.Split(KYCDetail.ProviderRejectReason, ",")
						reason = labelSlice[0]
						dcsMps, err := r.GetAllDecisionMaps(ctx)
						if err != nil {
							logc.Errorw(ctx, "r.GetAllDecisionMaps error", logc.Field("err", err))
						} else {
							k := fmt.Sprintf("%d-%s", ntf.Provider, reason)
							if cmt, ok := dcsMps[k]; ok {
								reason = cmt
							}
						}
					}

					txt := fmt.Sprintf(
						"Lv2 authentication failure notification (times:%d)\n1. UID: %d\n2. Sumsub Applicant_ID: %s\n3. Country: %s\n4. Failure time: %s\n5. Reason: %s",
						rjtTimes,
						ntf.MemberID,
						ntf.AccountID,
						KYCDetail.Country,
						tm.Format("2006-02-01 15:04:05"),
						reason,
					)
					_ = r.poaTeller.KycAlarm(ctx, txt, errors.New("Poa EEA alarm"))
				}
			}
		}
	}
	metrics.AddMetrics(nil, monitor.END)
	return true, nil
}

func (r *repo) sendLevel2StateComplete(ctx context.Context, ntf *kae_dto.KYCEventNotify) error {
	// 通知
	ntfMsg := &kae_dto.KYCEventNotify{
		EventType:    ekycadmin.KYCAdminAction(kae_dto.Level2StateComplete),
		MemberID:     ntf.MemberID,
		BrokerID:     ntf.BrokerID,
		Provider:     kyc.KYCProvider(int32(ntf.Provider) - 1), // 消费入口时候+1，-1为恢复之前的provider
		Level:        KycLv2,
		EventTimeMs:  time.Now().UnixNano() / int64(time.Millisecond),
		ResultDetail: ntf.ResultDetail,
		AccountID:    ntf.AccountID,
		RecordID:     ntf.RecordID,
	}
	msg, err := json.Marshal(ntfMsg)
	if err != nil {
		logc.Errorw(ctx, "json.Marshal error", logc.Field("err", err))
		return err
	}
	// 发送消息
	logc.Infow(ctx, "SendLevel2StateComplete", logc.Field("msg", string(msg)))

	return r.kafka.Send(ctx, &kafka.Message{
		Topic: KycActionTopic,
		Value: msg,
	})
}

func (r *repo) appendAccountID(ctx context.Context, record *o_kyc_action_state.Record, ntf *kae_dto.KYCEventNotify) {
	// 如果是 Jumio 去查一下
	if ntf.Provider == kyc.KYCProvider_PROVIDER_JUMIO && record.RecordId == "" {
		userAccountID, userDataID, err := r.GetJumioWrkFlwID(ctx, ntf.MemberID)
		if err != nil {
			bzap.LogError(r.logger.With(zap.Int64("memberID", ntf.MemberID)), "GetJumioWrkFlwID error", err)
			return
		}
		record.AccountId = userAccountID
		record.RecordId = userDataID
	}
	record.SelectDocType = ntf.SelectDocType
	if ntf.SelectCountry != "" {
		record.SelectCountry = helper.CountryCode2Alpha3(ntf.SelectCountry)
	}
}

func (r *repo) businessIDHandler(record *o_kyc_action_state.Record, ntf *kae_dto.KYCEventNotify) {
	if ntf.BusinessID != "" {
		record.BusinessId = ntf.BusinessID
	} else if record.BusinessId == "" && ntf.GetSerialType() == kp_dto.SerialTypeSecondary {
		record.BusinessId = constant.BusinessTypeCard.String()
	} else if record.BusinessId == "" {
		record.BusinessId = constant.BusinessTypeGlobal.String()
	}
}

func (r *repo) actionCreateHandler(ctx context.Context, record *o_kyc_action_state.Record, ntf *kae_dto.KYCEventNotify) {
	record.Action = int8(ekycadmin.KYCAdminAction_ACTION_CREATE)
	record.Channel = int8(ntf.Provider)
	record.Level = int8(ntf.Level)
	if ntf.RecordID != "" && record.RecordId == "" {
		record.RecordId = ntf.RecordID
	}
	if ntf.AccountID != "" && record.AccountId == "" {
		record.AccountId = ntf.AccountID
	}
	if ntf.ExternalUID != "" && record.ExternalUid == "" {
		record.ExternalUid = ntf.ExternalUID
	}
	// 供应商回调时间戳（ms）
	record.ProviderCreateAt = ntf.EventTimeMs
}

func (r *repo) actionPendingHandler(ctx context.Context, record *o_kyc_action_state.Record, ntf *kae_dto.KYCEventNotify) {
	record.Action = int8(ekycadmin.KYCAdminAction_ACTION_PENDING)
	record.Channel = int8(ntf.Provider)
	record.Level = int8(ntf.Level)
	if ntf.RecordID != "" && record.RecordId == "" {
		record.RecordId = ntf.RecordID
	}
	if ntf.AccountID != "" && record.AccountId == "" {
		record.AccountId = ntf.AccountID
	}
	if ntf.ExternalUID != "" && record.ExternalUid == "" {
		record.ExternalUid = ntf.ExternalUID
	}
	// 供应商回调时间戳（ms）
	record.ProviderCreateAt = ntf.EventTimeMs
}

func (r *repo) actionCompleteHandler(ctx context.Context, record *o_kyc_action_state.Record, ntf *kae_dto.KYCEventNotify, detail *kp_dto.KYCAdminDetail) error {
	metrics := monitor.NewMetrics("actionCompleteHandler", "")
	logc.Debugw(ctx, "actionCompleteHandler", logc.Field("ntf", ntf), logc.Field("KYCDetail", detail))
	ctx = logx.ContextWithFields(ctx, logx.Field("func", "actionCompleteHandler"))

	// 判断是不是新鲜事件：事件事件>=记录记录时间
	isFresh := ntf.EventTimeMs >= record.ProviderCreateAt
	logc.Infow(ctx, "actionCompleteHandler", logc.Field("isFresh", isFresh), logc.Field("ntf.EventTimeMs", ntf.EventTimeMs),
		logc.Field("record.ProviderCreateAt", record.ProviderCreateAt))
	if isFresh || ntf.ResultDetail.FinalDecision == 1 {
		record.Action = int8(ekycadmin.KYCAdminAction_ACTION_COMPLETE)
		record.Level = int8(ntf.Level)
		// 供应商回调时间戳（ms）
		record.ProviderCreateAt = ntf.EventTimeMs
		if ntf.ResultDetail.ReviewTimeMs > 0 {
			record.ReviewTime = ntf.ResultDetail.ReviewTimeMs
		} else {
			record.ReviewTime = ntf.EventTimeMs
		}
		if ntf.ResultDetail.FinalDecision > 0 {
			record.Decision = int8(ntf.ResultDetail.FinalDecision)
		}
		if len(ntf.ResultDetail.FinalReason) > 0 {
			record.RejectReason = ntf.ResultDetail.FinalReason
		}
		if record.Decision == 1 {
			record.RejectReason = ""
		}
	} else {
		r.logger.Info("Old event", zap.Any("notify", ntf), zap.Any("record", record))
		// 过期事件仍然更新 RejectReason
		if len(ntf.RawResponse.One) > 0 && len(ntf.ResultDetail.FinalReason) > 0 {
			record.RejectReason = ntf.ResultDetail.FinalReason
		}
	}
	if ntf.RecordID != "" && record.RecordId == "" {
		record.RecordId = ntf.RecordID
	}
	if ntf.AccountID != "" && record.AccountId == "" {
		record.AccountId = ntf.AccountID
	}
	if ntf.ExternalUID != "" && record.ExternalUid == "" {
		record.ExternalUid = ntf.ExternalUID
	}

	if detail == nil {
		return nil
	}
	record.ProviderDecision = detail.ProviderDecision
	record.ProviderRejectReason = detail.ProviderRejectReason
	record.DocType = detail.DocType
	record.DocSubtype = detail.DocSubtype
	// 需要加密
	dnEn, errDn := helper.EnAES256CodeString(r.secHub, detail.DocNumber)
	if errDn != nil {
		logc.Errorw(ctx, "EnAES256CodeString detail.DocNumber error", logc.Field("err", errDn), logc.Field("detail", detail))
		return errDn
	}
	if len(dnEn) > 255 {
		logc.Warnw(ctx, "AdminFieldTooLong", logc.Field("DocNumber", detail.DocNumber))
		dnEn, _ = helper.EnAES256CodeString(r.secHub, "")
	}
	record.DocNumber = dnEn
	record.Country = detail.Country
	record.Nationality = detail.Nationality
	record.AddressCountry = detail.AddressCountry
	// 需要加密
	fnEn, errFn := helper.EnAES256CodeString(r.secHub, detail.FirstName)
	if errFn != nil {
		logc.Errorw(ctx, "EnAES256CodeString detail.FirstName error", logc.Field("err", errFn), logc.Field("detail", detail))
		return errFn
	}
	if len(fnEn) > 255 {
		logc.Warnw(ctx, "AdminFieldTooLong", logc.Field("FirstName", detail.FirstName))
		fnEn, _ = helper.EnAES256CodeString(r.secHub, "")
	}
	record.FirstName = fnEn
	// 需要加密
	lnEn, errLn := helper.EnAES256CodeString(r.secHub, detail.LastName)
	if errLn != nil {
		logc.Errorw(ctx, "EnAES256CodeString detail.LastName error", logc.Field("err", errLn), logc.Field("detail", detail))
		return errLn
	}
	if len(lnEn) > 255 {
		logc.Warnw(ctx, "AdminFieldTooLong", logc.Field("LastName", detail.LastName))
		lnEn, _ = helper.EnAES256CodeString(r.secHub, "")
	}
	record.LastName = lnEn
	// 需要加密
	dobEn, errDob := helper.EnAES256CodeString(r.secHub, detail.Dob)
	if errDob != nil {
		logc.Errorw(ctx, "EnAES256CodeString detail.Dob error", logc.Field("err", errDob), logc.Field("detail", detail))
		return errDob
	}
	record.Dob = dobEn
	// DocDetail 需要加密
	docJSON, _ := jsoniter.MarshalToString(detail.DocDetail)
	docJSONStr, err := helper.EnAES256CodeString(r.secHub, docJSON)
	if err != nil {
		logc.Errorw(ctx, "EnAES256CodeString detail.DocDetail error", logc.Field("err", err), logc.Field("detail", detail))
		return err
	}
	record.DocDetail = docJSONStr

	record.IssuingDate = timeutil.ToTimeYmd(detail.IssuingDate)
	record.ExpireDate = timeutil.ToTimeYmd(detail.ExpireDate)

	// FaceDetail 需要加密
	faceJSON, err := jsoniter.MarshalToString(detail.FaceDetail)
	if err != nil {
		r.logger.Warn("detail.FaceDetail MarshalToString", zap.String("RecordID", ntf.RecordID), zap.Error(err))
	}
	faceJSONStr, err := helper.EnAES256CodeString(r.secHub, faceJSON)
	if err != nil {
		r.logger.Warn("faceJSON EnAES256CodeString", zap.String("RecordID", ntf.RecordID), zap.Error(err))
	}
	r.logger.Debug("actionCompleteHandler", zap.String("faceJSONStr", faceJSONStr), zap.String("RecordID", ntf.RecordID))
	record.FaceDetail = faceJSONStr

	// AddressDetail 需要加密
	addrJSON, _ := jsoniter.MarshalToString(detail.AddressDetail)
	addrJSONStr, errAd := helper.EnAES256CodeString(r.secHub, addrJSON)
	if errAd != nil {
		logc.Errorw(ctx, "EnAES256CodeString detail.AddressDetail error", logc.Field("err", errAd), logc.Field("detail", detail))
		return errAd
	}
	record.AddressDetail = addrJSONStr

	// ClientDetail 需要先解开，再附上 source
	if record.ClientDetail != "" {
		clientDtl := &kp_dto.KYCAdminClientDetail{}
		err := jsoniter.UnmarshalFromString(record.ClientDetail, clientDtl)
		if err == nil {
			clientDtl.Source = detail.ClientDetail.Source
		}
		if ntf.ClientDetail.IP != "" {
			clientDtl.ClientIP = ntf.ClientDetail.IP
		}
		if ntf.ClientDetail.Device != "" {
			clientDtl.From = ntf.ClientDetail.Device
		}
		clientJSON, _ := jsoniter.MarshalToString(clientDtl)
		record.ClientDetail = clientJSON
	}
	// 修正时间
	if detail.ReviewAtE3 > 0 {
		record.ReviewTime = detail.ReviewAtE3
		record.ProviderCreateAt = detail.ReviewAtE3
	}

	metrics.AddMetrics(nil, monitor.END)
	return nil
}

func (r *repo) actionResetHandler(ctx context.Context, record *o_kyc_action_state.Record, ntf *kae_dto.KYCEventNotify) {
	if ntf.IsAmend { // KYC 后台自己修改结果
		record.Action = int8(ekycadmin.KYCAdminAction_ACTION_COMPLETE)
	} else {
		record.Action = int8(ekycadmin.KYCAdminAction_ACTION_RESET)
	}
	record.SiteId = int16(ntf.BrokerID)
	record.Channel = int8(ntf.Provider)
	if ntf.Level != 0 {
		record.Level = int8(ntf.Level)
	}
	if ntf.RecordID != "" && record.RecordId == "" {
		record.RecordId = ntf.RecordID
	}
	if ntf.AccountID != "" && record.AccountId == "" {
		record.AccountId = ntf.AccountID
	}
	if ntf.ExternalUID != "" && record.ExternalUid == "" {
		record.ExternalUid = ntf.ExternalUID
	}
	// 供应商回调时间戳（ms）
	record.ProviderCreateAt = ntf.EventTimeMs
	if ntf.ResultDetail.ReviewTimeMs > 0 {
		record.ReviewTime = ntf.ResultDetail.ReviewTimeMs
	} else {
		record.ReviewTime = ntf.EventTimeMs
	}
	if ntf.ResultDetail.FinalDecision > 0 {
		record.Decision = int8(ntf.ResultDetail.FinalDecision)
	} else {
		record.Decision = int8(kp_dto.ReviewResultReject)
	}
	if len(ntf.ResultDetail.FinalReason) > 0 {
		record.RejectReason = ntf.ResultDetail.FinalReason
	}
}

// Get KYC detail from provider
func (r *repo) getKYCDetail(
	ctx context.Context,
	idempotentID string,
	ntf *kae_dto.KYCEventNotify,
) (*kp_dto.KYCAdminDetail, []*kp_dto.KYCActionImage, error) {
	var (
		detail      *kp_dto.KYCAdminDetail
		actionImgs  []*kp_dto.KYCActionImage
		firstPoi    *kp_dto.Identify
		err, errPoi error
	)
	metrics := monitor.NewMetrics("getKYCDetail", "")
	pvd := kycprovider.NewProvider(r.logger, kp_dto.KYCProvider(ntf.Provider), r.teller)
	if ntf.From == kae_dto.EventSourceMigrate && ntf.Level == 1 && ntf.LevelName == kp_dto.Level1.String() {
		firstPoi, errPoi = r.GetFirstPoi(ctx, ntf.MemberID)
		logc.Infow(ctx, "r.GetFirstPoi info", logc.Field("firstPoi", firstPoi), logc.Field("notify", ntf))
		if errPoi != nil {
			logc.Errorw(ctx, "r.GetFirstPoi error", logc.Field("err", errPoi))
		} else if firstPoi != nil {
			if string(firstPoi.Type) == kp_dto.DocumentTypeResidence.String() {
				firstPoi.Type = kp_dto.DocumentTypeResidence
			}
		}
	}
	detail, actionImgs, err = pvd.GetKYCAdminDetail(ctx, idempotentID, ntf, firstPoi)
	logc.Infow(
		ctx,
		"pvd.GetKYCAdminDetail info",
		logc.Field("detail", detail),
		logc.Field("actionImgs", actionImgs),
	)
	if err != nil {
		metrics.AddMetrics(err, "GetKYCAdminDetail")
		logc.Errorw(ctx, "pvd.GetKYCAdminDetail error", logc.Field("err", err))
		return nil, nil, err
	}
	if detail == nil {
		logc.Warnw(ctx, "pvd.GetKYCAdminDetail nil", logc.Field("notify", ntf))
		return nil, nil, nil
	}

	metrics.AddMetrics(nil, monitor.END)

	return detail, actionImgs, nil
}

func (r *repo) appendMemberKycApplicant(ctx context.Context, msg *kae_dto.KYCEventNotify, sdkDetail *kp_dto.KYCAdminDetail) bool {
	logc.Infow(ctx, "appendMemberKycApplicant begin", logc.Field("msg", msg), logc.Field("sdkDetail", sdkDetail))
	if sdkDetail == nil {
		logc.Warnw(ctx, "get sdkDetail info empty", logc.Field("msg", msg))
		return false
	}
	req := &kycpb.UpdateApplicantInfoRequest{
		MemberId: msg.MemberID,
		Address: &kycpb.GetKycProfileResponse_Address{
			Country:     sdkDetail.AddressDetail.GetCountry(),
			State:       sdkDetail.AddressDetail.GetState(),
			City:        sdkDetail.AddressDetail.GetCity(),
			FullAddress: sdkDetail.AddressDetail.GetFullAddress(),
			PostalCode:  sdkDetail.AddressDetail.GetPostalCode(),
		},
	}
	logc.Infow(ctx, "get request info", logc.Field("req", req))
	if !r.kycSvc.UpdateApplicantInfo(ctx, req) {
		logc.Errorw(ctx, "kycSvc.UpdateApplicantInfo failed")
		return false
	}
	logc.Infow(ctx, "appendMemberKycApplicant end")
	return true
}
